// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: support agent between domain
// feature: provide query interfaces
// feature: Split history as an independent common module
// feature: discovery support client and server
// ------------------------------------------------------------------

#include <discovery/endpoint/EDPServerListeners.hpp>

#include <elog/Log.hpp>
#include <history/WriterHistory.h>
#include <ertps/rtps/history/ReaderHistory.h>
#include <edds/rtps/reader/RTPSReader.h>

#include <discovery/endpoint/EDPServer.hpp>
#include <discovery/participant/PDPServer.hpp>
// #include <rtps/reader/BaseReader.hpp>
#include <edds/rtps/writer/StatefulWriter.h>

namespace evbs {
namespace ertps {
namespace rtps {

PDPServer* EDPServerPUBListener::get_pdp() {
    return sedp_->get_pdp();
}

EDPServerPUBListener::EDPServerPUBListener(EDPServer* sedp) : sedp_(sedp) {}

void EDPServerPUBListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                 const CacheChange_t* const change_in) {
    logDebug(RTPS_EDP_LISTENER, "");
    logDebug(RTPS_EDP_LISTENER, "------------------ EDP PUB SERVER LISTENER START ------------------");
    logDebug(RTPS_EDP_LISTENER,
             "-------------------- " << sedp_->mp_RTPSParticipant->getGuid() << " --------------------");

    // Create a new change from the one received
    CacheChange_t* change = (CacheChange_t*)change_in;
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    logDebug(RTPS_EDP_LISTENER, "EDP Server PUB Message received: " << change_in->instanceHandle);

    // DATA(w)s should have key
    if (!computeKey(change)) {
        elogWarning(RTPS_EDP_LISTENER, "Received change with no Key");
    }

    // Get EDP publications' reader history
    ReaderHistory* reader_history = reader->getHistory();

    // Related_sample_identity could be lost in message delivered, so we set as sample_identity
    // An empty related_sample_identity could lead into an empty sample_identity when resending this msg
    if (change->write_params.related_sample_identity() == SampleIdentity::unknown()) {
        change->write_params.related_sample_identity(change->write_params.sample_identity());
    }

    // Reset the internal CacheChange_t union.
    change->writer_info.next = nullptr;
    change->writer_info.previous = nullptr;
    change->writer_info.num_sent_submessages = 0;

    // DATA(w) case: new writer or updated information about an existing writer
    if (change->kind == ALIVE) {
        // Use a lambda to capture this explicitly and allow dynamic qos setting
        WriterAddedCallback writer_added_callback = [this](RTPSReader* reader, CacheChange_t* change,
                                                           vbsutil::xmlparser::WriterQos& qos, bool take_locator) {
            this->continue_with_writer(reader, change, qos, take_locator);
        };
        // Note: add_writer_from_change() removes the change from the EDP publications' reader history, but it does not
        // return it to the pool
        add_writer_from_change(reader, reader_history, change, sedp_, false, writer_added_callback);

        // DATA(w) case: Retrieve the topic after creating the WriterProxyData (in add_writer_from_change()). This way,
        // not matter whether the DATA(w) is a new one or an update, the WriterProxyData exists, and so the topic can be
        // retrieved

        // Stop and wait for callback in case of TypeLookupService needed time to process the types
        return;
    }
    // DATA(Uw) case
    else {
        logDebug(RTPS_EDP_LISTENER, "Disposed Remote Writer, removing...");

        // DATA(Uw) case: Retrieve the topic before removing the WriterProxyData. We need it to add the DATA(Uw) to the
        // database
        GUID_t auxGUID = iHandle2GUID(change->instanceHandle);
        std::string topic_name;
        std::string type_name;

        (void)get_writer_proxy_topic_name(auxGUID, topic_name, type_name);

        // Remove WriterProxy data information
        get_pdp()->removeWriterProxyData(auxGUID);

        // Removing change from history, not returning the change to the pool, since the ownership will be yielded to
        // the database
        reader_history->remove_change(change, false);

        const vbsutil::xmlparser::WriterQos qos;
        notify_discoverydatabase(std::move(topic_name), std::move(type_name), reader, change, qos);
    }
}

bool EDPServerPUBListener::get_writer_proxy_topic_name(GUID_t auxGUID, std::string& topic_name,
                                                       std::string& type_name) {
    auto temp_writer_data = get_pdp()->get_temporary_writer_proxies_pool().get();
    if (get_pdp()->lookupWriterProxyData(auxGUID, *temp_writer_data)) {
        topic_name = temp_writer_data->topicName().to_string();
        type_name = temp_writer_data->typeName().to_string();
        return true;
    } else {
        // This is a normal case in server redundancy scenarios
        logDebug(RTPS_EDP_LISTENER, "Writer Proxy Data missing for change " << auxGUID);
    }
    return false;
}

void EDPServerPUBListener::notify_discoverydatabase(std::string topic_name, std::string type_name, RTPSReader* reader,
                                                    CacheChange_t* change, const WriterQos& qos, bool take_locator) {
    // Notify the DiscoveryDataBase if it is enabled already
    // In case it is not enable, the change should not be updated or released because it is been
    // updated from a backup
    if (!get_pdp()->discovery_db().backup_in_progress()) {
        if (topic_name.size() > 0 &&
            get_pdp()->discovery_db().update(change, std::move(topic_name), qos, std::move(type_name), take_locator)) {
            // From here on, the discovery database takes ownership of the CacheChange_t. Henceforth there are no
            // references to the CacheChange_t.
            // Ensure processing time for the cache by triggering the Server thread (which process the updates
            get_pdp()->awake_routine_thread();
        } else {
            // If the database doesn't take the ownership, then return the CacheChante_t to the pool.
            reader->releaseCache(change);
        }
    }

    logDebug(RTPS_EDP_LISTENER,
             "-------------------- " << sedp_->mp_RTPSParticipant->getGuid() << " --------------------");
    logDebug(RTPS_EDP_LISTENER, "------------------ EDP PUB SERVER LISTENER END ------------------");
    logDebug(RTPS_EDP_LISTENER, "");
}

void EDPServerPUBListener::continue_with_writer(RTPSReader* reader, CacheChange_t* change,
                                                vbsutil::xmlparser::WriterQos& qos, bool take_locator) {
    std::string topic_name;
    std::string type_name;
    (void)get_writer_proxy_topic_name(iHandle2GUID(change->instanceHandle), topic_name, type_name);
    notify_discoverydatabase(std::move(topic_name), std::move(type_name), reader, change, qos, take_locator);
}

PDPServer* EDPServerSUBListener::get_pdp() {
    return sedp_->get_pdp();
}

EDPServerSUBListener::EDPServerSUBListener(EDPServer* sedp) : sedp_(sedp) {}

void EDPServerSUBListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                                 const CacheChange_t* const change_in) {
    logDebug(RTPS_EDP_LISTENER, "");
    logDebug(RTPS_EDP_LISTENER, "------------------ EDP SUB SERVER LISTENER START ------------------");
    logDebug(RTPS_EDP_LISTENER,
             "-------------------- " << sedp_->mp_RTPSParticipant->getGuid() << " --------------------");

    // Create a new change from the one received
    CacheChange_t* change = (CacheChange_t*)change_in;
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);

    logDebug(RTPS_EDP_LISTENER, "EDP Server SUB Message received: " << change_in->instanceHandle);

    // DATA(r)s should have key
    if (!computeKey(change)) {
        elogWarning(RTPS_EDP_LISTENER, "Received change with no Key");
    }

    // Related_sample_identity could be lost in message delivered, so we set as sample_identity
    // An empty related_sample_identity could lead into an empty sample_identity when resending this msg
    if (change->write_params.related_sample_identity() == SampleIdentity::unknown()) {
        change->write_params.related_sample_identity(change->write_params.sample_identity());
    }

    // Reset the internal CacheChange_t union.
    change->writer_info.next = nullptr;
    change->writer_info.previous = nullptr;
    change->writer_info.num_sent_submessages = 0;

    // Get BaseReader
    // BaseReader* base_reader = BaseReader::downcast(reader);
    // Get EDP subscriptions' reader history
    ReaderHistory* reader_history = reader->getHistory();

    // DATA(r) case: new reader or updated information about an existing reader
    if (change->kind == ALIVE) {
        ReaderAddedCallback reader_added_callback = [this](RTPSReader* reader, CacheChange_t* change,
                                                           vbsutil::xmlparser::ReaderQos& qos, bool take_locator,
                                                           uint32_t queue_id) {
            this->continue_with_reader(reader, change, qos, take_locator, queue_id);
        };
        // Note: add_reader_from_change() removes the change from the EDP subscriptions' reader history, but it does not
        // return it to the pool
        add_reader_from_change(reader, reader_history, change, sedp_, false, reader_added_callback);

        // DATA(w) case: Retrieve the topic after creating the ReaderProxyData (in add_reader_from_change()). This way,
        // not matter whether the DATA(r) is a new one or an update, the ReaderProxyData exists, and so the topic can be
        // retrieved

        // Stop and wait for callback in case of TypeLookupService needed time to process the types
        return;
    }
    // DATA(Ur) case
    else {
        // REMOVE WRITER FROM OUR READERS:
        logDebug(RTPS_EDP_LISTENER, "Disposed Remote Reader, removing...");

        // DATA(Uw) case: Retrieve the topic before removing the ReaderProxyData. We need it to add the DATA(Ur) to the
        // database
        GUID_t auxGUID = iHandle2GUID(change->instanceHandle);
        std::string topic_name;
        std::string type_name;

        (void)get_reader_proxy_topic_name(auxGUID, topic_name, type_name);

        // Remove ReaderProxy data information
        get_pdp()->removeReaderProxyData(auxGUID);

        // Removing change from history, not returning the change to the pool, since the ownership will be yielded to
        // the database
        reader_history->remove_change(change, false);

        const vbsutil::xmlparser::ReaderQos qos;
        notify_discoverydatabase(std::move(topic_name), std::move(type_name), reader, change, qos);
    }
}

bool EDPServerSUBListener::get_reader_proxy_topic_name(GUID_t auxGUID, std::string& topic_name,
                                                       std::string& type_name) {
    auto temp_reader_data = get_pdp()->get_temporary_reader_proxies_pool().get();
    if (get_pdp()->lookupReaderProxyData(auxGUID, *temp_reader_data)) {
        topic_name = temp_reader_data->topicName().to_string();
        type_name = temp_reader_data->typeName().to_string();
        return true;
    } else {
        elogWarning(RTPS_EDP_LISTENER, "Reader Proxy Data missing for change " << auxGUID);
    }
    return false;
}

void EDPServerSUBListener::notify_discoverydatabase(std::string topic_name, std::string type_name, RTPSReader* reader,
                                                    CacheChange_t* change, const vbsutil::xmlparser::ReaderQos& qos,
                                                    bool take_locator, uint32_t queue_id) {
    // Notify the DiscoveryDataBase if it is enabled already
    // In case it is not enable, the change should not be updated or released because it is been
    // updated from a backup
    if (!get_pdp()->discovery_db().backup_in_progress()) {
        if (topic_name.size() > 0 && get_pdp()->discovery_db().update(change, std::move(topic_name), qos,
                                                                      std::move(type_name), take_locator, queue_id)) {
            // From here on, the discovery database takes ownership of the CacheChange_t. Henceforth there are no
            // references to the CacheChange_t.
            // Ensure processing time for the cache by triggering the Server thread (which process the updates
            get_pdp()->awake_routine_thread();
        } else {
            // If the database doesn't take the ownership, then return the CacheChante_t to the pool.
            reader->releaseCache(change);
        }
    }

    logDebug(RTPS_EDP_LISTENER,
             "-------------------- " << sedp_->mp_RTPSParticipant->getGuid() << " --------------------");
    logDebug(RTPS_EDP_LISTENER, "------------------ EDP SUB SERVER LISTENER END ------------------");
    logDebug(RTPS_EDP_LISTENER, "");
}

void EDPServerSUBListener::continue_with_reader(RTPSReader* reader, CacheChange_t* change,
                                                vbsutil::xmlparser::ReaderQos& qos, bool take_locator,
                                                uint32_t queue_id) {
    std::string topic_name;
    std::string type_name;
    (void)get_reader_proxy_topic_name(iHandle2GUID(change->instanceHandle), topic_name, type_name);
    notify_discoverydatabase(std::move(topic_name), std::move(type_name), reader, change, qos, take_locator, queue_id);
}

} /* namespace rtps */
}  // namespace ertps
} /* namespace evbs */
